package org.zjt.flink.stream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.BigIntSerializer;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.DeltaTrigger;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-10-10 上午11:08
 * @see
 */
public class WindowWordCount2 {

    public static void main(String[] args) throws Exception {
        realTimeWindowWordCount2();
    }

    /**
     * size 没size个事件就统计一次
     * @throws Exception
     */
    public static void realTimeWindowWordCount2() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Random random = new Random();

        KeyedStream<Tuple2<String, Integer>, Tuple> tuple2TupleKeyedStream = env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");

                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {

            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value, random.nextInt(1000));
            }
        }).keyBy(0);

        TypeSerializer<Tuple2<String, Integer>> serializer = tuple2TupleKeyedStream.getType().createSerializer(env.getConfig());
        tuple2TupleKeyedStream.window(GlobalWindows.create())

                // evictor is window size       (计算的时间内的数据)
                .evictor(TimeEvictor.of(Time.of(10, TimeUnit.SECONDS)))

                // trigger is slide size        (什么时候执行，并输出)
                .trigger(DeltaTrigger.of(2, new DeltaFunction<Tuple2<String, Integer>>() {
                    @Override
                    public double getDelta(Tuple2<String, Integer> oldDataPoint, Tuple2<String, Integer> newDataPoint) {
                        return oldDataPoint.f1 - newDataPoint.f1;
                    }
                },serializer)).maxBy(1).print();

        env.execute();
    }



    public static void realTimeWindowWordCount() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {

            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).timeWindow(Time.seconds(30)).sum(1).print();

        env.execute();
    }


    /**
     * timeWindow(Time size, Time slide)   size：窗口时间大小     slide:抽样的频率，slide秒滑动一下，并做统计。
     * @throws Exception
     */
    public static void intermittentlyWindowWordCount() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {

            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).timeWindow(Time.seconds(30),Time.seconds(1)).sum(1).print();
        env.execute();
    }



    public static void tenCountWindowWordCount() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {

            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).countWindow(10).sum(1).print();
        env.execute();
    }



    public static void tenCountWindowWordCount2() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                List<String> arrays = Arrays.asList("zhang", "wang", "zhang", "li");
                Random random = new Random();
                for (int j = 0; j < 100; j++) {
                    TimeUnit.SECONDS.sleep(1);
                    int i = random.nextInt(arrays.size());
                    ctx.collect(arrays.get(i));
                }
            }

            @Override
            public void cancel() {

            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value,1);
            }
        }).keyBy(0).countWindow(10,4).sum(1).print();
        env.execute();
    }
}
